// Package parallel Description: 描述
package parallel

import (
	"github.com/zeromicro/go-zero/core/fx"
	"github.com/zeromicro/go-zero/core/mr"
)

// Finish 开放式的并行执行器，自定义包装函数（并发期有函数错误返回，会终止未执行的执行体执行）
func Finish(fns ...func() error) error {
	number := len(fns)
	// 防止协程过多，与go-zero框架默认值保持一致
	if number > 16 {
		number = 16
	}

	return mr.MapReduceVoid(func(source chan<- interface{}) {
		for _, fn := range fns {
			source <- fn
		}
	}, func(item interface{}, writer mr.Writer[interface{}], cancel func(error)) {
		fn := item.(func() error)
		if err := fn(); err != nil {
			cancel(err)
		}
	}, func(pipe <-chan interface{}, cancel func(error)) {
		for range pipe {
		}
	}, mr.WithWorkers(number))
}

// FinishGroup 开放式的并行执行器(所有包装函数同时执行)
func FinishGroup(fns ...func()) {
	fx.Parallel(fns...)
}

// Parallel 并发器(并行中的执行体出错，会终止未执行的执行体执行)，
// 一般使用场景是对列表数据进行并发操作（比如数据的回填、数据处理等）
func Parallel(r int, f func(index int) error) error {
	if r == 0 {
		return nil
	}

	fns := make([]func() error, 0, r)
	for i := 0; i < r; i++ {
		fns = append(fns, warpFn(i, f))
	}

	return Finish(fns...)
}

// UnrelatedParallel 非关联的并发器(执行体之间不受影响，同时执行)
func UnrelatedParallel(r int, f func(index int)) {
	if r == 0 {
		return
	}

	fns := make([]func(), 0, r)
	for i := 0; i < r; i++ {
		fns = append(fns, warpFnNoRet(i, f))
	}

	fx.Parallel(fns...)
}

// SafetyParallel 协程安全的并发器(并行中的执行体出错，会终止未执行的执行体执行)，
// 通过revRet统一接受结果，不要在f里面操作结果写map之类不安全的类型，应该在revRet里面写map之类不安全的类型
func SafetyParallel(r int, f func(index int) (interface{}, error), revRet func(interface{})) error {
	if r == 0 {
		return nil
	}

	fns := make([]func() (interface{}, error), 0, r)
	for i := 0; i < r; i++ {
		fns = append(fns, warpSafetyFn(i, f))
	}

	return finishSafetyFn(fns, revRet)
}
